package com.dtsheep.rxjava.stduy.observable;

import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.schedulers.Schedulers;

/**
 * UnsafeCreateDemo
 *
 * <p>
 * </p>
 *
 * @author chuanbai
 * @date 2024/5/17
 */
public class UnsafeCreateDemo {

    public void unsafeCreate01(){
        Observable<String> observable = Observable.unsafeCreate((Subscriber<? super String> s) -> {
            s.onNext(sendMessage());
            s.onNext("111");
            s.onNext("测试");
            s.onNext("小猫");
            s.onCompleted();
        }).subscribeOn(Schedulers.io());

        try {
            System.out.println("====休眠中====");
            Thread.sleep(5000);
            System.out.println("====休眠结束====");
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        // 阻塞发射数据
        String first = observable.toBlocking().first();
        System.out.println("阻塞发射数据" + first);

        System.out.println("开始订阅");
        // 订阅
        Subscription subscription = observable.subscribe(
                event1 -> System.out.println(event1),
                event2 -> System.out.println(event2),
                () -> System.out.println("发射空事件")
        );

    }
    public String sendMessage(){
        return "火箭发射";
    }

    public static void main(String[] args) {
        UnsafeCreateDemo unsafeCreateDemo = new UnsafeCreateDemo();
        unsafeCreateDemo.unsafeCreate01();
    }
}
